using System;
using System.Collections;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Linq.Expressions;
using System.Reflection;
using System.Text;
using Apache.Cassandra;
using Cassandraemon.Builder;
using Cassandraemon.Connection;
using Cassandraemon.FullText;
using Cassandraemon.Operators;
using Cassandraemon.Util;
using Thrift.Collections;
using Thrift.Protocol;
using Thrift.Transport;


namespace Cassandraemon
{
	[DebuggerDisplay("{ClientHostName}")]
	public class CassandraContext : IDisposable
	{
		public string Host { get { return this._ConnectionConfig.Hosts.First(); } }

		public int Port { get { return this._ConnectionConfig.Port; } }

#if SET_KEYSPACE
        private string _Keyspace;
        public string Keyspace { get { return this._Keyspace ?? this._ConnectionConfig.Keyspace; } }
        [Obsolete("Use Keyspace instead")]
		public string KeySpace { get { return this._Keyspace ?? this._ConnectionConfig.Keyspace; } }
#else
        public string Keyspace { get { return this._ConnectionConfig.Keyspace; } }
        [Obsolete("Use Keyspace instead")]
        public string KeySpace { get { return this._ConnectionConfig.Keyspace; } }
#endif

		public ConsistencyLevel ConsistencyLevel { get; set; }
		
		public bool HasError { get; private set; }
		
		private bool _IsDisposed = false;
		
		public CassandraQueryable<CassandraEntity<Column>> Column 
		{ 
			get { return new CassandraQueryable<CassandraEntity<Column>>(this); } 
		}
		
		public CassandraQueryable<CassandraEntity<SuperColumn>> SuperColumn 
		{ 
			get { return new CassandraQueryable<CassandraEntity<SuperColumn>>(this); } 
		}
		
		public CassandraQueryable<CassandraEntity<CounterColumn>> CounterColumn 
		{ 
			get { return new CassandraQueryable<CassandraEntity<CounterColumn>>(this); } 
		}
		
		public CassandraQueryable<CassandraEntity<CounterSuperColumn>> CounterSuperColumn 
		{ 
			get { return new CassandraQueryable<CassandraEntity<CounterSuperColumn>>(this); } 
		}
		
		public CassandraQueryable<CassandraEntity<List<Column>>> ColumnList 
		{ 
			get { return new CassandraQueryable<CassandraEntity<List<Column>>>(this); } 
		}
		
		public CassandraQueryable<CassandraEntity<List<SuperColumn>>> SuperColumnList 
		{ 
			get { return new CassandraQueryable<CassandraEntity<List<SuperColumn>>>(this); } 
		}

		public CassandraQueryable<CassandraEntity<List<CounterColumn>>> CounterColumnList 
		{ 
			get { return new CassandraQueryable<CassandraEntity<List<CounterColumn>>>(this); } 
		}
		
		public CassandraQueryable<CassandraEntity<List<CounterSuperColumn>>> CounterSuperColumnList 
		{ 
			get { return new CassandraQueryable<CassandraEntity<List<CounterSuperColumn>>>(this); } 
		}

		#region Constructors/Destructors

        public CassandraContext()
            : this(CassandraConnectionConfig.Default)
        {
        }

        #region Constructors from CassandraConnectionConfig

        public CassandraContext(CassandraConnectionConfig config)
        {
            if (config == null)
                config = CassandraConnectionConfig.Default;

            this._ConnectionConfig = config;
            this.ConsistencyLevel = config.ConsistencyLevel;

            HasError = false;

            var pool = ConnectionPool.GetOrCreate(config);
            _Pool = pool;
            Client = pool.Borrow();

            this.OnCreated();
        }

        public CassandraContext(IDictionary properties)
            : this(new CassandraConnectionConfig(properties))
        {
        }

        public CassandraContext(CassandraConnectionConfig config, IDictionary properties)
            : this(new CassandraConnectionConfig(config, properties))
        {
        }

        #endregion

        #region Legacy Constructors

        public CassandraContext(string host, int port, string keySpace) 
			: this(new CassandraConnectionConfig(host, port, keySpace))
		{
		}
		
		public CassandraContext(string host, int port, string keySpace, ConsistencyLevel consistencyLevel)
            : this(new CassandraConnectionConfig(host, port, keySpace), new CassandraConnectionConfigBuilder {
                ConsistencyLevel = consistencyLevel,
            })
		{
		}

		public CassandraContext(string host, int port, string keySpace, ConsistencyLevel consistencyLevel, bool isFramed)
            : this(new CassandraConnectionConfig(host, port, keySpace), new CassandraConnectionConfigBuilder {
                ConsistencyLevel = consistencyLevel,
                IsFramed = isFramed,
            })
		{
		}
	
		public CassandraContext(string host, 
			                    int port, 
			                    string keySpace, 
			                    ConsistencyLevel consistencyLevel, 
			                    bool isFramed,
			                    int retryCount,
			                    int timeoutMillisecond, 
			                    int minPoolCountPerHost, 
			                    int maxPoolCountPerHost)
            : this(new CassandraConnectionConfig(host, port, keySpace), new CassandraConnectionConfigBuilder {
                ConsistencyLevel = consistencyLevel,
                RetryCount = retryCount,
                IsFramed = isFramed,
                Timeout = TimeSpan.FromMilliseconds(timeoutMillisecond),
                MinPoolCountPerHost = minPoolCountPerHost,
                MaxPoolCountPerHost = maxPoolCountPerHost,
            })
		{
		}

        #endregion

        #region Constructors from ConnectionPool

        public CassandraContext(ConnectionPool pool)
			: this(pool, null)
		{
		}

		public CassandraContext(ConnectionPool pool, CassandraConnectionConfig config)
		{
            if (pool == null)
            {
                throw new ArgumentNullException("pool");
            }
			if (config == null)
			{
				config = pool.ConnectionConfig;
			}
			else if (!CassandraConnectionConfig.PoolKeyComparer.Equals(config, pool.ConnectionConfig))
			{
				throw new ArgumentException("ConnectionConfig does not match to the pool");
			}

            this._ConnectionConfig = config;
			this.ConsistencyLevel = config.ConsistencyLevel;

            HasError = false;

			_Pool = pool;
			Client = pool.Borrow();

            this.OnCreated();
		}

        #endregion

        ~CassandraContext()
		{
			Dispose(false);
		}

		#endregion

        #region CassandraConfig

        readonly CassandraConnectionConfig _ConnectionConfig;

        public CassandraConnectionConfig ConnectionConfig
        {
            get { return _ConnectionConfig; }
        }

		public int RetryCount { get { return this._ConnectionConfig.RetryCount; } }

        #endregion

        #region Pools

		private ConnectionPool _Pool;

		public ConnectionPool Pool
		{
			get
			{
				var pool = this._Pool;
				if (pool != null && !pool.IsDisposed)
				{
					return pool;
				}
                pool = ConnectionPool.GetOrDefault(this._ConnectionConfig);
                if (pool != null && !pool.IsDisposed)
                {
					return pool;
				}
				return null;
			}
		}

		#endregion

        #region Client

        public Cassandra.Client Client { get; private set; }
        public string ClientHostName { get { return this.Client.HostName(); } }

        public void ExecuteClient(Action<Cassandra.Client> action)
        {
            this.Operate(() => action(this.Client));
        }

        public T ExecuteClient<T>(Func<Cassandra.Client, T> action)
        {
            return this.Operate(() => action(this.Client));
        }

        #endregion

        #region Execute Queries

		internal object Execute(Expression expression, bool isEnumerable) 
		{
			if(this._IsDisposed) throw new ObjectDisposedException(this.GetType().FullName);
			
			Expression cassandraExpression = GetCassandraExpression (expression);
			Type cassandraType = cassandraExpression.Type;
			
			object obj;
			
			if(cassandraType.Equals(typeof(int)))
			{
				obj = Operate(() => BuildCassandraOperator<int>(cassandraExpression).Operate());
				return obj;
			}
			else if(cassandraType.Equals(typeof(long)))
			{
				obj = Operate(() => BuildCassandraOperator<long>(cassandraExpression).Operate());
				return obj;
			}
			else if(cassandraType.Name == "Dictionary`2")
			{
				obj = Operate(() => BuildCassandraOperator<Dictionary<byte[], int>>(cassandraExpression).Operate());
				
				Type keyType = cassandraType.GetGenericArguments()[0];
				object dic = Activator.CreateInstance(cassandraType);
				foreach(var kv in (Dictionary<byte[], int>)obj)
				{
					dic.GetType().GetMethod("Add").Invoke(dic, new object[]{MappingUtil.MappingValueType(keyType, kv.Key), kv.Value});
				}
				return dic;
			}
			else if(cassandraType.Equals(typeof(IQueryable<CassandraEntity<Column>>)))
			{
				obj = Operate(() => BuildCassandraOperator<Column>(cassandraExpression).Operate());
				return Execute<Column>(expression, cassandraExpression, isEnumerable, obj as List<CassandraEntity<Column>>);
			}
			else if(cassandraType.Equals(typeof(IQueryable<CassandraEntity<SuperColumn>>)))
			{
				obj = Operate(() => BuildCassandraOperator<SuperColumn>(cassandraExpression).Operate());
				return Execute<SuperColumn>(expression, cassandraExpression, isEnumerable, obj as List<CassandraEntity<SuperColumn>>);
			}
			else if(cassandraType.Equals(typeof(IQueryable<CassandraEntity<CounterColumn>>)))
			{
				obj = Operate(() => BuildCassandraOperator<CounterColumn>(cassandraExpression).Operate());
				return Execute<CounterColumn>(expression, cassandraExpression, isEnumerable, obj as List<CassandraEntity<CounterColumn>>);
			}
			else if(cassandraType.Equals(typeof(IQueryable<CassandraEntity<CounterSuperColumn>>)))
			{
				obj = Operate(() => BuildCassandraOperator<CounterSuperColumn>(cassandraExpression).Operate());
				return Execute<CounterSuperColumn>(expression, cassandraExpression, isEnumerable, obj as List<CassandraEntity<CounterSuperColumn>>);
			}
			else if(cassandraType.Equals(typeof(IQueryable<CassandraEntity<List<Column>>>)))
			{
				obj = Operate(() => BuildCassandraOperator<List<Column>>(cassandraExpression).Operate());
				return Execute<List<Column>>(expression, cassandraExpression, isEnumerable, obj as List<CassandraEntity<List<Column>>>);
			}
			else if(cassandraType.Equals(typeof(IQueryable<CassandraEntity<List<SuperColumn>>>)))
			{
				obj = Operate(() => BuildCassandraOperator<List<SuperColumn>>(cassandraExpression).Operate());
				return Execute<List<SuperColumn>>(expression, cassandraExpression, isEnumerable, obj as List<CassandraEntity<List<SuperColumn>>>);
			}
			else if(cassandraType.Equals(typeof(IQueryable<CassandraEntity<List<CounterColumn>>>)))
			{
				obj = Operate(() => BuildCassandraOperator<List<CounterColumn>>(cassandraExpression).Operate());
				return Execute<List<CounterColumn>>(expression, cassandraExpression, isEnumerable, obj as List<CassandraEntity<List<CounterColumn>>>);
			}
			else if(cassandraType.Equals(typeof(IQueryable<CassandraEntity<List<CounterSuperColumn>>>)))
			{
				obj = Operate(() => BuildCassandraOperator<List<CounterSuperColumn>>(cassandraExpression).Operate());
				return Execute<List<CounterSuperColumn>>(expression, cassandraExpression, isEnumerable, obj as List<CassandraEntity<List<CounterSuperColumn>>>);
			}
			
			return null;
		}
		
		private Expression GetCassandraExpression (Expression expression)
		{
			if(!(expression is MethodCallExpression))
			{
				throw new InvalidOperationException("No query over the data source was specified.");
			}
			
			ExpressionAnalyzer analyzer = new ExpressionAnalyzer();
			analyzer.Analyze(expression);
			Expression cassandraExpression = analyzer.CassandraExpression;
			
			if(!(cassandraExpression is MethodCallExpression))
			{
				throw new InvalidOperationException("No query over the data source was specified.");
			}
			
			return cassandraExpression;
		}

		private ICassandraOperator<T> BuildCassandraOperator<T> (Expression cassandraExpression)
		{
			CassandraOperatorBuilder<T> builder
				= new CassandraOperatorBuilder<T>(this.Keyspace, this.ConsistencyLevel);
			
			CassandraOperatorCondition<T> condition = builder.Build(cassandraExpression);
			
			ICassandraOperator<T> oper = condition.CreateOperator();
			oper.Context = this;
			oper.Client = Client;
			
			return oper;
		}

		
		private object Execute<T>(Expression expression, Expression cassandraExpression, bool isEnumerable, List<CassandraEntity<T>> entityList)
		{
			IQueryProvider provider = null;
			
			var queryable = entityList == null ? null : entityList.AsQueryable();
			if(queryable != null) provider = queryable.Provider;
			
			Expression newExpression = ExpressionReplacer.Replace(expression, cassandraExpression, Expression.Constant(queryable));
			
			if(isEnumerable)
			{
				return provider.CreateQuery(newExpression);
			}
			else
			{
				return provider.Execute(newExpression);
			}
		}
		

        #endregion

        #region Insert

        public void InsertOnSubmit(
            string columnFamily,
            object key,
            Column column)
        {
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), column);
        }

        public void InsertOnSubmit(
            string columnFamily,
            object key,
            SuperColumn column)
        {
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), column);
        }

        public void InsertOnSubmit(
            string columnFamily,
            object key,
            CounterColumn column)
        {
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), column);
        }

        public void InsertOnSubmit(
            string columnFamily,
            object key,
            CounterSuperColumn column)
        {
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), column);
        }

		public void InsertOnSubmit(
			string columnFamily,
			object key,
			ColumnOrSuperColumn column)
		{
			this.ValidateNotDisposed();
			this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), column);
		}

        public void InsertOnSubmit(
            string columnFamily,
            object key,
            IEnumerable<Column> columns)
        {
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), columns);
        }

        public void InsertOnSubmit(
            string columnFamily,
            object key,
            IEnumerable<SuperColumn> columns)
        {
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), columns);
        }

        public void InsertOnSubmit(
            string columnFamily,
            object key,
            IEnumerable<CounterColumn> columns)
        {
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), columns);
        }

        public void InsertOnSubmit(
            string columnFamily,
            object key,
            IEnumerable<CounterSuperColumn> columns)
        {
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), columns);
        }

		public void InsertOnSubmit(
			string columnFamily,
			object key,
			IEnumerable<ColumnOrSuperColumn> columns)
		{
			this.ValidateNotDisposed();
			this._MutationMap.AddInsertion(columnFamily, key.ToCassandraByte(), columns);
		}

		internal void InsertOnSubmit<T>(T entity)
		{
            this.ValidateNotDisposed();
            this._MutationMap.AddInsertion(entity);
		}

        #endregion

        #region Delete

		internal void DeleteOnSubmit(LambdaExpression expression)
		{
			if(this._IsDisposed) throw new ObjectDisposedException(this.GetType().FullName);

            this._MutationMap.AddDeletion(expression);
		}

        #endregion

        #region FullText

        public void InsertFullTextIndexOnSubmit(object document,
		                                     	IAnalyzer analyzer,
		                                     	string columnFamily)
		{
			object docId = GetDocumentID(document);
			
			Type type = document.GetType();
			
			PropertyInfo[] properties = type.GetProperties();
			
			foreach(var p in properties)
			{
				foreach(var a in p.GetCustomAttributes(true))
				{
					if(a is DocumentFieldAttribute)
					{
						var attribute = (DocumentFieldAttribute)a;
						string fieldName = string.IsNullOrEmpty(attribute.FieldName) ? p.Name : attribute.FieldName;
						
						string text = p.GetValue(document, null).ToString();
						
						var indexes = analyzer.MakeRegsitData(text);
						
						foreach(var key in indexes.Keys)
						{
							var superColumn = new SuperColumn()
									.SetName(fieldName)
									.SetColumn(docId, indexes[key]);
							
							var entity = new CassandraEntity<SuperColumn>()
									.SetColumnFamily(columnFamily)
									.SetKey(key)
									.SetData(superColumn);
									
							InsertOnSubmit(entity);
						}
					}
				}
			}
		}
		
		public void DeleteFullTextIndexOnSubmit(object document,
		                                     	IAnalyzer analyzer,
		                                     	string columnFamily)
		{
			object docId = GetDocumentID(document);
			
			Type type = document.GetType();
			
			PropertyInfo[] properties = type.GetProperties();
			
			foreach(var p in properties)
			{
				foreach(var a in p.GetCustomAttributes(true))
				{
					if(a is DocumentFieldAttribute)
					{
						var attribute = (DocumentFieldAttribute)a;
						string fieldName = string.IsNullOrEmpty(attribute.FieldName) ? p.Name : attribute.FieldName;
						
						string text = p.GetValue(document, null).ToString();
						
						var indexes = analyzer.MakeRegsitData(text);
						
						foreach(var key in indexes.Keys)
						{
							Expression<Func<CassandraEntity<SuperColumn>, bool>> lambda = x =>
							               x.ColumnFamily == columnFamily &&
							               x.Key == key.ToCassandraByte() &&
							               x.SuperColumn == fieldName.ToCassandraByte() &&
							               x.Column == docId.ToCassandraByte();
							               
							DeleteOnSubmit(lambda);
							
							// somehow this need
							SubmitChanges();
						}
					}
				}
			}
		}
		
		private object GetDocumentID(object document)
		{
			object docId = null;
			
			Type type = document.GetType();
			
			PropertyInfo[] properties = type.GetProperties();
			
			foreach(var p in properties)
			{
				foreach(var a in p.GetCustomAttributes(true))
				{
					if(a is DocumentIdAttribute)
					{
						if(docId != null) throw new ArgumentException("More than one DocumentID is set.");
						docId = p.GetValue(document, null);
					}
				}
			}
			
			return docId;
        }

        #endregion
        
		#region CQL
		
		public CqlResult ExecuteCqlQuery(string query)
		{
			return ExecuteCqlQuery(query.ToCassandraByte());
		}
		
		public CqlResult ExecuteCqlQuery(byte[] query)
		{
			return ExecuteCqlQuery(query, Compression.NONE);
		}
		
		public CqlResult ExecuteCqlQuery(string query, Compression compression)
		{
			return ExecuteCqlQuery(query.ToCassandraByte(), compression);
		}
		
		public CqlResult ExecuteCqlQuery(byte[] query, Compression compression)
		{
			return Operate(() => Client.execute_cql_query(query, compression));
		}
		
		#endregion
		
        #region Mutations and SubmitChanges

        private Dictionary<byte[], Dictionary<string, List<Mutation>>> _MutationMap
            = new Dictionary<byte[], Dictionary<string, List<Mutation>>>(new ByteArrayEqualityComparer());

        public Dictionary<byte[], Dictionary<string, List<Mutation>>> MutationMap
        {
            get { return _MutationMap; }
        }

		public int MutationsCount
		{
			get { return this._MutationMap.CountMutations(); }
		}

		public void SubmitChanges()
		{
			if(this._IsDisposed) throw new ObjectDisposedException(this.GetType().FullName);
			
            var mutations = this._MutationMap;
            if (mutations.Count == 0)
            {
                return;
            }

            mutations.SetTimeStamp();

			Operate(() =>
			{
				var sw = Stopwatch.StartNew();
                try {
    				Client.batch_mutate(this._MutationMap, this.ConsistencyLevel);
                } catch (Exception err) {
                    this.OnExecuted(new CassandraExecuteEventArgs(err,
                        sw.Elapsed, this.Client, "batch_mutate", null));
                    throw;
                }
                this.OnExecuted(new CassandraExecuteEventArgs(
                    this.MutationsCount,
                    sw.Elapsed, this.Client, "batch_mutate", null));
			});


            // Clear Mutations only on success.
            // If batch_mutate fails, it should throw and Mutations should be kept for retry.

            this._MutationMap.Clear();
		}

        /// <summary>
        /// Removes <see cref="Mutation"/> that are redundant before sending to the server.
        /// This method reads data from the server to decide if a mutation is redundant or not,
        /// so it can have negative impacts.
        /// </summary>
		public void RemoveMutationsIfNoChanges(Func<int, bool> progress)
        {
			this._MutationMap.RemoveIfNoChange(this, progress);
        }

		public void RemoveMutationsIfNoChanges()
		{
			this.RemoveMutationsIfNoChanges(null);
		}

        #endregion

		public void Login(Dictionary<string, string> credentials)
		{
			Operate(() => Client.login(new AuthenticationRequest{Credentials = credentials}));
		}
		
#if SET_KEYSPACE
        // This method is deprecated because this can change default keyspace of a connection,
        // and can cause troubles like:
        // * Default keyspace changes on fail over
        // * Can contradict with ConnectionPool.Keyspace
        // See http://cassandraemon.codeplex.com/workitem/13
		public void SetKeyspace(string keyspace)
		{
			Operate(() => Client.set_keyspace(keyspace));
			this._Keyspace = keyspace;
		}
#endif

        public string DescribeClusterName()
		{
			return Operate(() => Client.describe_cluster_name());
		}
		
		public List<KsDef> DescribeKeySpaces()
		{
			return Operate(() => Client.describe_keyspaces());
		}
		
		public KsDef DescribeKeySpace()
		{
			return DescribeKeySpace(this.Keyspace);
		}
		
		public KsDef DescribeKeySpace(string keySpace)
		{
			return Operate(() => Client.describe_keyspace(keySpace));
		}
			
		public Dictionary<string, List<string>> DescribeSchemaVersions()
		{
			return Operate(() => Client.describe_schema_versions());
		}
		
		public List<TokenRange> DescribeRing()
		{
			return DescribeRing(this.Keyspace);
		}
		
		public List<TokenRange> DescribeRing(string keySpace)
		{
			return Operate(() => Client.describe_ring(keySpace));
		}
		
		public string DescribePartitioner()
		{
			return Operate(() => Client.describe_partitioner());
		}
		
		public string DescribeSnitch()
		{
			return Operate(() => Client.describe_snitch());
		}
		
		public List<string> DescribeSplits(string columnFamilyName, string startToken, string endToken, int keysPerSplit)
		{
			return Operate(() => Client.describe_splits(columnFamilyName, startToken, endToken, keysPerSplit));
		}
		
		public string DescribeVersion()
		{
			return Operate(() => Client.describe_version());
		}
		
		public string SystemAddKeyspace(KsDef ksDef)
		{
			return Operate(() => Client.system_add_keyspace(ksDef));
		}
		
		public string SystemUpdateKeyspace(KsDef ksDef)
		{
			return Operate(() => Client.system_update_keyspace(ksDef));
		}
		
		public string SystemDropKeyspace(string keyspace)
		{
			return Operate(() => Client.system_drop_keyspace(keyspace));
		}
		
		public string SystemAddColumnFamily(CfDef cfDef)
		{
			return Operate(() => Client.system_add_column_family(cfDef));
		}
		
		public string SystemUpdateColumnFamily(CfDef cfDef)
		{
			return Operate(() => Client.system_update_column_family(cfDef));
		}
		
		public string SystemDropColumnFamily(string columnFamily)
		{
			return Operate(() => Client.system_drop_column_family(columnFamily));
		}
		
		public void Truncate(string columnFamily)
		{
			Operate(() => Client.truncate(columnFamily));
		}
		
		private void Operate(Action action)
		{
			Operate<bool>(() => {action(); return true;});
		}
		
		internal TResult Operate<TResult>(Func<TResult> func)
		{
			if(this._IsDisposed) throw new ObjectDisposedException(this.GetType().FullName);
			
			var retryCount = RetryCount;

			// RetryCount = -1, the default value, means (#nodes * 2),
			// so that every node has a chance to retry.
			// See http://cassandraemon.codeplex.com/workitem/17

			if (retryCount < 0)
			{
				var pool = this.Pool;
				retryCount = pool.Nodes.Count * 2 - 1;
				if (retryCount < 0)
				{
					retryCount = 1;
				}
			}
			
			while(retryCount-- >= 0)
			{
				try
				{
					return func();
				}
				catch (Exception err)
				{
					TraceError("{2}: RetryCount={1}: {0}",
                        err,
                        retryCount,
                        this.ClientHostName);
					if (retryCount >= 0 &&
						(err is TimedOutException ||
						 err is UnavailableException ||
						 err is TTransportException ||
						 err is IOException))
					{
						RecoverError(err);
						continue;
					}
					HasError = true;
					throw;
				}
			}
			
			return default(TResult);
		}
		
		private void RecoverError(Exception err)
		{
            var pool = ConnectionPool.GetOrCreate(this._ConnectionConfig);
			pool.ErrorReturn(Client, err);
            Client = null;

			Client = pool.Borrow();
		}

        #region Close/Dispose

		public void Close()
		{
			Dispose();
		}
		
		public void Dispose ()
		{
			Dispose(true);
			
			GC.SuppressFinalize(this);
		}
		
		private void Dispose(bool disposing)
		{
			if(!this._IsDisposed)
			{
                if (Client != null)
                {
                    var pool = this.Pool;
                    if (pool != null && !pool.IsDisposed)
                    {
                        try
                        {
                            pool.Return(Client);
                            Client = null;
                        }
                        catch (ObjectDisposedException)
                        {
                        }
                    }

                    if (Client != null)
                    {
                        Client.Close();
                    }
                    Client = null;
                }
				
				_MutationMap = null;

				_IsDisposed = true;

                var handler = this.Disposed;
                if (handler != null)
                {
                    handler(this, EventArgs.Empty);
                }
			}
		}

        void ValidateNotDisposed()
        {
            if (this._IsDisposed) throw new ObjectDisposedException(this.GetType().FullName);
        }

        #endregion

        #region Events

        void OnCreated()
        {
            var handler = Created;
            if (handler != null)
            {
                handler(this, EventArgs.Empty);
            }
        }

        internal void OnExecuted(CassandraExecuteEventArgs e)
        {
            var handler = this.Executed;
            if (handler != null)
            {
                handler(this, e);
            }

#if TRACE
            var trace = TraceSource;
            var err = e.Exception;
            if (err != null)
            {
                var data = e.TraceDataArray;
                TraceFormatData(data);
                var msg = string.Concat(
                    err.GetType().Name, ": ",
                    err.Message, ": ",
                    string.Join(", ", data.Select(o => Convert.ToString(o)).ToArray()));
                trace.TraceEvent(TraceEventType.Error, TraceOperationId, msg);
            }
            else if (trace.Switch.ShouldTrace(TraceEventType.Verbose))
            {
                var data = e.TraceDataArray;
                TraceFormatData(data);
                trace.TraceData(TraceEventType.Verbose, TraceOperationId, data);
            }
#endif
        }

        public static event EventHandler Created;
        public event EventHandler Disposed;
        public event EventHandler<CassandraExecuteEventArgs> Executed;

        #endregion

        #region Diagnostics/Trace/Errors

        const string TraceSourceName = "Cassandraemon";
        const int TraceOperationId = 1969;

		public static readonly TraceSource TraceSource =
			new TraceSource(
				TraceSourceName,
				SourceLevels.Information);

		[Conditional("TRACE")]
		internal static void TraceError(string format, params object[] args)
		{
			TraceSource.TraceEvent(TraceEventType.Error, 0, format, args);
		}

		[Conditional("TRACE")]
		internal static void TraceWarning(string format, params object[] args)
		{
			TraceSource.TraceEvent(TraceEventType.Warning, 0, format, args);
		}

		[Conditional("TRACE")]
		internal static void TraceInformation(string format, params object[] args)
		{
			TraceSource.TraceEvent(TraceEventType.Information, 0, format, args);
		}

		[Conditional("TRACE")]
		internal static void Fail(string message, string detailMessage)
		{
			foreach (TraceListener listener in TraceSource.Listeners)
			{
				listener.Fail(message, detailMessage);
			}
		}

		[Conditional("TRACE")]
		internal static void Assert(bool condition, string message, string detailMessage)
		{
			if (!condition)
			{
				return;
			}

			foreach (TraceListener listener in TraceSource.Listeners)
			{
				listener.Fail(message, detailMessage);
			}
		}

        static void TraceFormatData(object[] data)
        {
            for (int i = 0; i < data.Length; i++)
            {
                var d = data[i];
                if (d == null)
                {
                    continue;
                }

                ICollection items = d as ICollection;
                if (items != null)
                {
                    data[i] = items.Count;
                    continue;
                }
            }
        }

		#endregion
	}
}
